package com.netty.websocket;

//import cn.netty.spring.JdbcUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.topsoft.common.MiscHelper;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class WebSocketHandler extends SimpleChannelInboundHandler<Object> {
	private int overLengthIsCompress=1000;//消息超过多长就压缩
	//	public boolean isErrClose = false;//出现错误 是否关闭连接，默认不关闭
//	public Logger logger = LogManager.getLogger("[netty]"+getClass().getName());
	public JSONObject parameters;
	public String queue;
	public Channel channel;
	public NettySession nettySession;
	public QueueType queueType;
	public Boolean isConfirm2Producer=true;    //是否将确认消息优先返回给原始生产者，找不到原始生产者，则含返回给该队列的其他生产者
	public Boolean isSendIsConfirm=true;       //是否服务器一发送，就记录为已消费确认【默认false，此时需要消费者发送确认信息，才记录为已消费确认】
	public Boolean isServerRecv2Producer=false;//则表示服务器收到消息，是否返回一条消息给生产者

	private String maxIdString="";
	private GroupManager groupManager=null;
	public JSONObject errJsonMsg=null;


	private static ConcurrentHashMap<String, JSONObject> mapSendMsg = new ConcurrentHashMap();//消息中有__code，已经发送过的下次不再发送
	private static ConcurrentHashMap<String, NettySession> mapNettySession = new ConcurrentHashMap();
	private static ConcurrentHashMap<ChannelId,String> channelIds=new ConcurrentHashMap();
	private static int maxId=1;
	private static int recvCount=0;

	//NioSocketChannel 是接口 channel的实现类
	public void onOpen() throws Exception{ }
	public void onMessage(JSONObject msgJSON) throws Exception{ }
	public void onClosed() throws Exception{ }
	public void onError(Throwable throwable) throws Exception{ }

	//接收到消息 每个消息的长度为 65536
	@Override
	protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
		if(mapSendMsg==null)     mapSendMsg      = new ConcurrentHashMap();
		if(mapNettySession==null)mapNettySession = new ConcurrentHashMap();
		if(channelIds==null)     channelIds      = new ConcurrentHashMap();

		recvCount++;
		if (recvCount>5000){
			recvCount=0;
			mapSendMsg.clear();
		}

		try {
			//接收到消息
			channel = ctx.channel();
			if (sendMsgErr()) return;//有错误消息，始终发送
			String msgString;
			ByteBuf byteBuf;
			byte[] bytes;
			if (msg instanceof TextWebSocketFrame){
				msgString=((TextWebSocketFrame)msg).text();
				if (msgString.startsWith("__gzip:")) {
					msgString=msgString.substring(7);
					msgString=MiscHelper.gunzip(msgString,"UTF-8");//解压
				}
			}else if (msg instanceof BinaryWebSocketFrame) {
				byteBuf = ((BinaryWebSocketFrame)msg).content();
				bytes = new byte[byteBuf.readableBytes()];
				byteBuf.readBytes(bytes);                    //二进制数据
				msgString= MiscHelper.gunzip(bytes,"UTF-8");//解决得到的字符串
			}else{
				log.warn("消息 queue=" + queue + " " + getSockId() + " channelRead0 读取消息类型 不存在 ");
				return;
			}

			JSONObject msgJson; String __code="";
			try {
				msgJson = JSONObject.parseObject(msgString);
				__code=msgJson.getString("__code");
			} catch (Exception e) {
				msgJson = new JSONObject(true);
				msgJson.put("msg", msgString);
			}
			if (channel.isOpen()) {
				log.debug("消息 queue={} {} length:{} {}  消息 {}",
						queue, getSockId(), msgString.length(), channel.remoteAddress(), MiscHelper.isNull(__code)?msgString:__code);
				onMessage(msgJson);
				sendMessage(msgJson);//广播消息，若消息指定了接收队列，则在指定的队列中广播，否则在当前队列中广播
			}else{
				log.warn("消息 queue=" + queue + " " + getSockId() + " 已关闭 消息 " + msgString + " ");
			}
		} catch (Exception e) {
			log.warn("消息 queue=" + queue + " " + getSockId() + " 出现错误，关闭连接 ");
			e.printStackTrace();
			ctx.close();
		}
	}
	String err="";
	public void afterAdd(Channel channel2) throws Exception{
		if (sendMsgErr())return;//有错误消息，始终发送

		String queueType2 =parameters.getString("queueType");
		if (queueType2==null || queueType2.equals("")){
			queueType2=QueueType.PC_MORE.name();
		}else if (!queueType2.equals(QueueType.PC_ONE.name()) && !queueType2.equals(QueueType.PC_MORE.name())
				&& !queueType2.equals(QueueType.C_ONE.name()) && !queueType2.equals(QueueType.C_MORE.name())
				&& !queueType2.equals(QueueType.P_ONE.name()) && !queueType2.equals(QueueType.P_MORE.name())){
			sendMsgByQueueTypeErr();
			return;
		}

		queueType=QueueType.valueOf(queueType2);
		maxIdString=MiscHelper.fillStr(""+(maxId++),3,"0");
		channel=channel2;
		nettySession=new NettySession(maxIdString,channel2,queue,queueType,parameters);
		mapNettySession.put(maxIdString,nettySession);
		channelIds.put(channel.id(),maxIdString);

//		String userId=parameters.getString("userId");
//		String sessionId=nettySession.getSessionId();
//		if ((userId==null || userId.equals("")) && sessionId!=null && sessionId.length()>2) {
//			JSONObject session=JdbcUtil.queryForMap("select UserId,Datasource from SecSessionInfo where SessionId='" + sessionId + "'");
//			if (session==null) {
//				err="sessionId 查无记录。";
//			}else{
//				userId = session.getString("UserId");
//				parameters.put("userId", userId);
//			}
//		}

		if (groupManager==null)groupManager = GroupManager.newOrFind(queue);
		groupManager.add(queueType,nettySession);

		sendMsgByOnlineOffline(true); //发送连接成功的消息
		sendMsgByMessageCache();//发送缓存消息

		onOpen();
	}

	public void sendMsgByMessageCache(){
		//发送缓存消息 开始********************************************************************************************************
		//有消费者用户加入，则发送生产者缓存的消息
		if (queueType==QueueType.C_ONE || queueType==QueueType.C_MORE){
			JSONObject msgJSON=MessageCache.newOrFind("P2C_"+queue).poll();//获得一个消息
			while(msgJSON!=null){
				channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(msgJSON)));
				msgJSON=MessageCache.newOrFind("P2C_"+queue).poll();//获得下一个消息
			}
		}
		//有生产者用户加入，则发送消费者缓存的消息
		if (queueType==QueueType.P_ONE || queueType==QueueType.P_MORE){
			JSONObject msgJSON=MessageCache.newOrFind("C2P_"+queue).poll();//获得一个消息
			while(msgJSON!=null){
				log.debug("C2P_"+queue+" 发送缓存消息"+msgJSON.toString());
				//luojp msgJSON必须要克隆一个新对象，消息才能发送成功 奇怪
				channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(msgJSON)));
				msgJSON=MessageCache.newOrFind("C2P_"+queue).poll();//获得下一个消息
			}
		}
		//发送缓存消息 完毕********************************************************************************************************
	}
	private void removeChannel(ChannelHandlerContext ctx)   {
		if (queue==null || queue.length()==0){
			Attribute<String> attrQueue= channel.attr(AttributeKey.valueOf("queue"));
			queue=attrQueue.get();
		}

		if (queue==null || queue.length()==0){
			queue=nettySession.getQueue();
		}
		mapNettySession.remove(maxIdString);
		channelIds.remove(channel.id());
		if (sendMsgErr()){//有错误消息，始终发送
			channel.close();
			return;
		}

		try{
			if (groupManager==null) groupManager = GroupManager.newOrFind(queue);
			if (groupManager!=null) groupManager.remove(queueType,nettySession);
			onClosed();
		}catch(Exception e){
			e.printStackTrace();
			ctx.close();
			channel.close();

		}
	}

	/**
	 * 发送消息
	 * @param recvQueue 接收消息的队列
	 * @param msgJSON   消息内容
	 */
	public void sendMessage(String recvQueue, JSONObject msgJSON) {
		if (recvQueue==null || recvQueue.equals("")|| msgJSON==null) return;
		GroupManager recvGroupManager = GroupManager.newOrFind(recvQueue);
		String __type=msgJSON.getString("__type");__type=__type==null?"":__type;
		String __code=msgJSON.getString("__code");
		JSONObject heads=msgJSON.getJSONObject("__heads");
		String pSockId=null;
		NettySession producerConfirmNettySession=null;
		if(__type.equals("confirm")){
			msgJSON.put("info","已消费");
			if(heads!=null) {
				heads.put("cSockId", getSockId());//原始消息消费者
				heads.put("cUserId", parameters.getString("userId"));//原始消息消费者
				pSockId=heads.getString("pSockId");pSockId="".equals(pSockId)?null:pSockId;
			}else{
				System.out.println("heads==null");
			}
			if (pSockId!=null) producerConfirmNettySession=mapNettySession.get(pSockId);//找到消息生产者

			//记录已消费确认过的消息
			if (__code!=null && __code.length()>0) {
				mapSendMsg.put(queue + "-"+__code, msgJSON);
			}
		}else{
			heads=heads==null?new JSONObject(MiscHelper.isDebug):heads;
			heads.put("pSockId",getSockId());
			heads.put("pUserId",nettySession.getUserd());
			heads.put("queue", queue);      //发送的队列
			heads.put("nowTime", MiscHelper.getDateTime());
			msgJSON.put("__heads",heads);

//			if (__code!=null && __code.length()>0 && mapSendMsg.containsKey(queue +"-"+ __code)){//假如是 已消费确认，则禁止发送
//				JSONObject msgErrJSON=new JSONObject(true);
//				msgErrJSON.put("__type","error");
//				msgErrJSON.put("number","E202");
//				msgErrJSON.put("info","当前消息已被消费过，不能多次发送");
//				msgErrJSON.put("__code",__code);
//				msgErrJSON.put("heads",heads);
//				log.debug("消息 queue=" + queue + " " + getSockId() + " __code=" + __code+"已经拦截");
//				//给当前生产者反馈错误
//				channel.writeAndFlush(new TextWebSocketFrame(msgErrJSON.toJSONString()));
//				return; //返回，禁止执行后面的逻辑
//			}
		}

		msgJSON.remove("__queue");
		String msgString=msgJSON.toJSONString();
		WebSocketFrame webSocketFrame;

		//消费后的确认消息  始终不压缩
		if(__type.equals("confirm")==false && msgString.length()>overLengthIsCompress){
			try {
				msgString= "__encode:"+URLEncoder.encode(msgString,"UTF-8");//转码，解决前段 解压后 中文的问题
			} catch (UnsupportedEncodingException e) {
				e.printStackTrace();
			}
			msgString="__gzip:"+MiscHelper.gzip2String(msgString,"UTF-8");
			webSocketFrame=new TextWebSocketFrame(msgString);
//			webSocketFrame=new BinaryWebSocketFrame(Unpooled.wrappedBuffer(msgString.getBytes()));
			//webSocketFrame=new BinaryWebSocketFrame(Unpooled.wrappedBuffer(MiscHelper.gzip2byte(msgString,"UTF-8")));
//			webSocketFrame=new BinaryWebSocketFrame(Unpooled.wrappedBuffer(msgString.getBytes()));

		}else {
			webSocketFrame=new TextWebSocketFrame(msgString);
		}

		if(isConfirm2Producer && producerConfirmNettySession!=null){
			//if(isConfirm2Producer && producerConfirmNettySession!=null && (queueType==QueueType.C_ONE || queueType==QueueType.PC_ONE)){
			//有找到消息生产者，则将消费后的确认消息 发送给它
			producerConfirmNettySession.channel().writeAndFlush(webSocketFrame);
		}else if (queueType.name().indexOf("C_")==0){
			//当前是消费者，给生产者发送消息
			boolean ret1=groupManager.writeAndFlush(QueueType.P_ONE, nettySession,webSocketFrame);
			boolean ret2=groupManager.writeAndFlush(QueueType.P_MORE,nettySession,webSocketFrame);

			if(ret1==false && ret2==false){//假如都没有成功发送消息
				MessageCache.newOrFind("C2P_"+queue).add(msgJSON);//缓存消息：C2P表示消息来源是消费者C,消息将发送给生产者P
			}
		}else{
			//当前是生产者，这里是给消费者发送消息
			boolean ret1=recvGroupManager.writeAndFlush(QueueType.PC_ONE, nettySession,webSocketFrame);
			boolean ret2=recvGroupManager.writeAndFlush(QueueType.PC_MORE,nettySession,webSocketFrame);
			boolean ret3=recvGroupManager.writeAndFlush(QueueType.C_ONE,  nettySession,webSocketFrame);
			boolean ret4=recvGroupManager.writeAndFlush(QueueType.C_MORE, nettySession,webSocketFrame);
			if(ret3==false && ret4==false) {//假如都没有成功发送消息
				MessageCache.newOrFind("P2C_"+queue).add(msgJSON);//缓存消息：P2C_ 表示消息来源是 生产者P,消息将发送给消费者C
			}
			if(isSendIsConfirm && __code!=null && __code.length()>0){//true，表示 服务器一发送，就记录为已消费确认
				mapSendMsg.put(queue + "-"+__code, msgJSON);//记录为已消费确认
			}
		}
	}

	public void sendMessage(String recvQueue, String msgString) {
		if (recvQueue==null || recvQueue.equals("")|| msgString==null) return;

		JSONObject msgJSON;
		try {
			msgJSON = JSONObject.parseObject(msgString);
		} catch (Exception e) {
			msgJSON = new JSONObject(MiscHelper.isDebug);
			msgJSON.put("msg", msgJSON);
		}

		sendMessage(recvQueue,msgJSON);
	}
	public void sendMessage(JSONObject msgJSON) {
		if (msgJSON==null) return;

		String recvQueue;
		if (msgJSON!=null){
			//客户端发送的消息 指定了 队列，则发送到指定的队列名  __queue 中
			String __queue=msgJSON.getString("__queue");
			recvQueue = (__queue!=null && __queue.length()>0)?__queue:queue;
		}else{
			recvQueue=queue;
		}
		sendMessage(recvQueue, msgJSON);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {//7
		if (cause instanceof IOException){

		}else {
			cause.printStackTrace();
			//log.info("异常 queue=" + queue + " " + getSockId() + " " + cause.getMessage());
		}
		onError(cause);

		super.exceptionCaught(ctx,cause);
		// 当出现异常就关闭连接
		ctx.close();
	}


//	public static String getSockId(Channel channel2)  {
//		return channelIds.get(channel2.id());
//	}

	public String getSockId()  {
		return channelIds.get(channel.id());
	}

	@Override
	public void handlerAdded(ChannelHandlerContext ctx) throws Exception {  // (2)
		channel = ctx.channel();
		Attribute<WebSocketHandler> attrWebSocketHandler= channel.attr(AttributeKey.valueOf("webSocketHandler"));
		attrWebSocketHandler.set(this);
	}


	@Override
	public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {  // (3)
		channel = ctx.channel();
		removeChannel(ctx);
		sendMsgByOnlineOffline(false);
		super.handlerRemoved(ctx);
	}

	//isOnline =true 上线， false下线
	private void sendMsgByOnlineOffline(boolean isOnline){
//	    if (nettySession.getQueueType()!=QueueType.PC_MORE){
//	        return ;//只有PC_MORE 才有 上下线用户的通知
//        }

		List listUsers=groupManager.getOnlineUsers();
		String userId=parameters.getString("userId");
		String info=(isOnline?"上线通知:":"下线通知:") +" 在线用户数="+listUsers.size()+" 连接数="+groupManager.getOnlineSocks()+" 用户信息: sockId="+getSockId()+" userId="+userId;
		if (err!=null && err.length()>2)info=err+info;
		JSONObject serverInfo = new JSONObject(true);
		serverInfo.put("__type",(isOnline?"online":"offline"));
		serverInfo.put("number",(isOnline?"S001":"S002"));
		serverInfo.put("userId",userId);//上下线的用户Id
		serverInfo.put("info",info);
		serverInfo.put("queue",queue);
		serverInfo.put("sockId",getSockId());//上下线的sockId
		serverInfo.put("queueType",queueType.name());
		if(parameters.containsKey("sessionId"))	serverInfo.put("sessionId",parameters.getString("sessionId"));
		if(parameters.containsKey("subSystem"))	serverInfo.put("subSystem",parameters.getString("subSystem"));

		serverInfo.put("remoteAddress",channel.remoteAddress().toString());
		if(isConfirm2Producer)serverInfo.put("isConfirm2Producer",isConfirm2Producer);
		if(isSendIsConfirm)serverInfo.put("isSendIsConfirm",isSendIsConfirm);

		//生成在线信息
		JSONObject online = new JSONObject(true);
		serverInfo.put("onlineQueueType",online);
		for (QueueType key:groupManager.keySet()) {
			BaseChannelGroup baseChannelGroup=groupManager.get(key);
			if (baseChannelGroup!=null){
				online.put(key.name(),""+baseChannelGroup.size());
			}
		}
		log.debug((isOnline?"上线":"下线") +" queue="+queue+" "+getSockId()+" "+channel.remoteAddress() +" parameters="+parameters.toString());

		serverInfo.put("onlineUsers",listUsers);
		serverInfo.put("nowTime", MiscHelper.getDateTime());

		groupManager.writeAndFlush(nettySession,new TextWebSocketFrame(serverInfo.toJSONString()));//给 非 自己的客户端发送
		if (isOnline) {
			//给 自己的客户端发送
			serverInfo.put("__type","success");
			serverInfo.put("info",info.replace("上线通知","连接成功" ));
			channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(serverInfo)));
		}
	}
	private void sendMsgByQueueTypeErr(){
		errJsonMsg = new JSONObject(true);
		errJsonMsg.put("__type","error");
		errJsonMsg.put("number","E200");
		errJsonMsg.put("info","连接地址url中参数 queueType="+parameters.getString("queueType")+" 的值不对");
		errJsonMsg.put("queue",queue);
		errJsonMsg.put("sockId",getSockId());//

		log.debug("参数queueType错误"+" queue="+queue+" "+getSockId()+" "+channel.remoteAddress() +" parameters="+parameters.toString());
		channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(errJsonMsg)));
	}

	public boolean sendMsgErr(){
		if (errJsonMsg!=null){//有错误消息，始终发送
			channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(errJsonMsg)));
			return true;
		}
		return false;
	}

//	/**
//	 * 是否服务器一发送，就记录为已消费确认【默认管理，此时需要消费者发送确认信息，才记录为已消费确认】
//	 * @param value
//	 */
//	public void setIsSendIsConfirm(boolean value){
//		this.isSendIsConfirm=value;
//	}


//	public static String toStringByQueueType(){
//		String str="";
//		for (QueueType value: QueueType.values()) {
//			str=value+","+str;
//		}
//		return str;
//	}
}